package cn.edu.flink.scala.tutorial.多流转换

import cn.edu.flink.scala.tutorial.source.ClickSource
import org.apache.flink.streaming.api.functions.co.CoProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object ConnectTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()

    val urlStream = env.addSource(new ClickSource).map(_.url)
    val timestampStream = env.addSource(new ClickSource).map(_.timestamp)

    val connetedStream: ConnectedStreams[String, Long] = urlStream.connect(timestampStream)

    val resultStream: DataStream[String] = connetedStream.process(new CoProcessFunction[String, Long, String] {
      override def processElement1(value: String, ctx: CoProcessFunction[String, Long, String]#Context, out: Collector[String]): Unit = {
        out.collect(value)
      }

      override def processElement2(value: Long, ctx: CoProcessFunction[String, Long, String]#Context, out: Collector[String]): Unit = {
        out.collect(value.toString)
      }
    })

    resultStream.print()

    env.execute("ConnectTest")
  }
}
